package com.ikonke.iot.alarm.es;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.Resource;

import com.ikonke.iot.alarm.domain.PushMsgDO;
import com.ikonke.iot.alarm.mapper.PushMsgDOMapper;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.noear.snack.ONode;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Repository;
import org.springframework.util.CollectionUtils;

import com.ikonke.iot.alarm.domain.AlarmStatDO;
import com.ikonke.iot.alarm.email.MailDO;
import com.ikonke.iot.alarm.mapper.AlarmStatDOMapper;
import com.ikonke.iot.alarm.service.MailService;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.*;
import io.netty.channel.EventLoopGroup;
import lombok.extern.slf4j.Slf4j;
import tk.mybatis.mapper.entity.Example;

/**
 * @Author: liull
 * @Description: es repository 1.每一小时去获取最近前一小时（2020-06-11 19:00:00 000-2020-06-11 19:59:59 999）的所有异常信息（NullPointerException/ClassCastException等） 处理成
 *               ccuid - Map -- 异常类型 - 出现的次数 并保存至mysql数据库 2.每隔五分钟获取前五分钟的所有异常数据 （1）.如果是新出现的异常进行告警 （2）.否则不做额外的操作
 * @Date: Created in 16:49 2020/6/5
 * @Modified By:
 */
@Slf4j
@Repository
public class EsRepository {

    /**
     * es特有的日期格式
     */
    public static final String ES_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss,SSS";

    private static Map<String, List<String>> syncLogMap;

    static {
        // 同步日志的顺序
        syncLogMap = new HashMap<>();
        List<String> trackerList = new ArrayList<>();
        trackerList.add("执行发送请求报文：执行后台任务");
        trackerList.add("reqTask start");
        trackerList.add("reqTask success published");
        trackerList.add("parse SYNC_INFO of reqTask complete, status=success");
        syncLogMap.put("RequestTrackerTask", trackerList);
        List<String> clientList = new ArrayList<>();
        clientList.add("主机同步完成,准备入库");
        clientList.add("主机同步完成,入库完成");
        syncLogMap.put("InternalSHClient", clientList);
    }

    private Map<Date, Integer> somedayErrorLogCount = new ConcurrentHashMap<>();

    @Resource
    private AlarmStatDOMapper alarmStatDOMapper;

    @Resource
    private PushMsgDOMapper pushMsgDOMapper;

    @Autowired
    private MailService mailService;

    @Autowired
    private RestHighLevelClient esClient;

    @Autowired
    @Qualifier("eventLoopGroup")
    private EventLoopGroup eventLoopGroup;

    /**
     * 获取mysql统计信息并发送邮件
     *
     * @param types 1 ccuid 2 opcode
     * @param statType 1 日 2周 2月
     */
    public void getErrorMsgAndSend(String statType, String... types) {
        for (String type : types) {
            // 统计昨天的数据
            if (statType.equals("1")) {
                // 获取昨天的日期
                Date yesterday = DateUtil.offsetDay(new Date(), -1).toJdkDate();
                Date dayBeforeYsterday = DateUtil.offsetDay(yesterday, -1).toJdkDate();
                // 统计昨天之前出现的所有异常类型
                List<AlarmStatDO> statDOS = alarmStatDOMapper
                        .selectStatCountYesterday(type, DateUtil.format(dayBeforeYsterday, DatePattern.NORM_DATE_PATTERN));
                // 昨天出现的异常
                List<AlarmStatDO> statDOS2 = alarmStatDOMapper.selectStatCount(type, DateUtil.format(yesterday, DatePattern.NORM_DATE_PATTERN));
                // 返回新出现的异常类型
                if (!CollectionUtils.isEmpty(statDOS)) {
                    Iterator<AlarmStatDO> iterator = statDOS2.iterator();
                    while (iterator.hasNext()) {
                        AlarmStatDO statDO = iterator.next();
                        for (int i = 0; i < statDOS.size(); i++) {
                            if (statDOS.get(i).equals(statDO)) {
                                iterator.remove();
                            }
                        }
                    }
                }
                if (CollUtil.isNotEmpty(statDOS2)) {
                    MailDO mailDO = new MailDO();
                    mailDO.setTitle("设备云平台昨天新出现的异常统计信息");
                    mailDO.setEmail("1442993929@qq.com,yuanhu@yeah.net");
                    // mailDO.setEmail("1442993929@qq.com");
                    Map<String, Object> map = new HashMap<>();
                    map.put("infos", statDOS2);
                    mailDO.setAttachment(map);
                    mailService.sendTemplateMail(mailDO);
                }
            } else if (statType.equals("2")) {
                // 暂不支持
            }
        }

    }

    public void statErrorMsg(Date date, String... types) {
        for (String type : types) {
            // 昨天出现的异常
            List<AlarmStatDO> statDOS = alarmStatDOMapper.selectStatCount(type, DateUtil.format(date, DatePattern.NORM_DATE_PATTERN));
            MailDO mailDO = new MailDO();
            mailDO.setTitle("设备云平台每日异常统计信息");
            mailDO.setEmail("1442993929@qq.com,yuanhu@yeah.net");
            // mailDO.setEmail("1442993929@qq.com");
            Map<String, Object> map = new HashMap<>();
            map.put("infos", statDOS);
            mailDO.setAttachment(map);
            mailService.sendTemplateMail(mailDO);
        }

    }

    /**
     * 初始化指定时间段出现的所有的异常类型，并统计ccuid,opcode对应每种异常次数
     */
    public void doInit() throws IOException {
        DateTime startTime = DateUtil.date();
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        SearchRequest searchRequest = new SearchRequest("filebeat*");
        searchRequest.scroll(scroll);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        Date test1 = DateUtil.parse("2020-12-26 10:23:07");
        Date test2 = DateUtil.parse("2020-12-26 11:00:07");
        searchSourceBuilder.query(QueryBuilders.boolQuery()
                // .must(QueryBuilders.matchAllQuery())
                // .must(QueryBuilders.matchQuery("message", "主机同步失败"))
                // .must(QueryBuilders.matchQuery("ccuid", "CCU_08069"))
                // 模糊查询
                .must(QueryBuilders.matchQuery("log_level", "ERROR"))
                // 范围查询
                .filter(QueryBuilders.rangeQuery("@timestamp")
                        // 当前时间18:00:00 124统计前一小时
                        // .gte(DateUtil.offsetHour(new Date(), -1)
                        // // 分钟置零
                        // .setField(DateField.MINUTE, 0)
                        // // 秒置零
                        // .setField(DateField.SECOND, 0)
                        // // 毫秒置零
                        // .setField(DateField.MILLISECOND, 0)
                        // .getTime())
                        // // .gte("2020-06-04 17:20:34,046")
                        // .lte(DateUtil.offsetHour(new Date(), -1)
                        // // 分钟置零
                        // .setField(DateField.MINUTE, 59)
                        // // 秒置零
                        // .setField(DateField.SECOND, 59)
                        // // 毫秒置零
                        // .setField(DateField.MILLISECOND, 999)
                        // .getTime()))
                        .gte(DateUtil.offsetMinute(startTime.toJdkDate(), -5)
                                // 秒置零
                                .setField(DateField.SECOND, 0)
                                // 毫秒置零
                                .setField(DateField.MILLISECOND, 0)
                                .getTime())
                        .lte(DateUtil.offsetMinute(startTime.toJdkDate(), -1)
                                // 秒置零
                                .setField(DateField.SECOND, 59)
                                // 毫秒置零
                                .setField(DateField.MILLISECOND, 999)
                                .getTime()))
        // 排序
        )
                .sort(SortBuilders.fieldSort("@timestamp").order(SortOrder.DESC));
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        // 统计 ccuid 对应的 异常类型 的 次数
        Map<String, Map<String, Integer>> statisticCcuId = new ConcurrentHashMap<>();
        // 统计 opcode 对应的 异常类型 的 次数
        Map<String, Map<String, Integer>> statisticOpcode = new ConcurrentHashMap<>();
        while (searchHits != null && searchHits.length > 0) {
            processSearchHits(searchHits, statisticCcuId, statisticOpcode);
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            searchResponse = esClient.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
        }
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = esClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        boolean succeeded = clearScrollResponse.isSucceeded();
        if (succeeded) {
            log.info("scroll 清理成功");
        } else {
            log.warn("scroll 清理失败");
        }
        List<AlarmStatDO> alarmStatDOList = new ArrayList<>();
        eventLoopGroup.next().execute(() -> {
            statisticCcuId.forEach((k, v) -> v.forEach((iK, iV) -> {
                Date now = new Date();
                // 定时汇总至mysql
                AlarmStatDO alarmStatDO = new AlarmStatDO();
                alarmStatDO.setCreateTime(now);
                alarmStatDO.setCreateDate(now);
                alarmStatDO.setCreateDateTime(now);
                // ccuid 1 opcode 2
                alarmStatDO.setType("1");
                alarmStatDO.setCount(iV.longValue());
                // 异常类型名
                alarmStatDO.setExTypeName(iK);
                // ccuid
                alarmStatDO.setMetrics(k);
                alarmStatDOList.add(alarmStatDO);
            }));
            statisticOpcode.forEach((k, v) -> v.forEach((iK, iV) -> {
                Date now = new Date();
                // 定时汇总至mysql
                AlarmStatDO alarmStatDO = new AlarmStatDO();
                alarmStatDO.setCreateTime(now);
                alarmStatDO.setCreateDate(now);
                alarmStatDO.setCreateDateTime(now);
                // ccuid 1 opcode 2
                alarmStatDO.setType("2");
                alarmStatDO.setCount(iV.longValue());
                // 异常类型名
                alarmStatDO.setExTypeName(iK);
                // opcode
                alarmStatDO.setMetrics(k);
                alarmStatDOList.add(alarmStatDO);
            }));
            // 批量插入
            if (CollectionUtils.isEmpty(alarmStatDOList)) {
                log.warn("没有统计到错误日志");
                return;
            }
            alarmStatDOMapper.insertList(alarmStatDOList);
        });
        DateTime endTime = DateUtil.date();
        log.info("统计最近5分钟的的数据完成，耗时：{}毫秒", DateUtil.between(startTime.toJdkDate(), endTime.toJdkDate(), DateUnit.MS));
    }

    /**
     * 处理SearchHits结果集
     *
     * @param statisticCcuId
     * @param statisticOpcode
     * @return
     */
    private SearchHit[] processSearchHits(SearchHit[] searchHits, Map<String, Map<String, Integer>> statisticCcuId,
            Map<String, Map<String, Integer>> statisticOpcode) {
        if (searchHits != null && searchHits.length > 0) {
            for (int i = 0; i < searchHits.length; i++) {
                Map<String, Integer> map = new ConcurrentHashMap<>();
                // 异常信息
                String message = (String) searchHits[i].getSourceAsMap().get("message");
                String log_level = (String) searchHits[i].getSourceAsMap().get("log_level");
                // System.err.println(log_level);
                String ccuId = (String) searchHits[i].getSourceAsMap().get("ccuid");
                String opcode = (String) searchHits[i].getSourceAsMap().get("opcode");
                String className = (String) searchHits[i].getSourceAsMap().get("class");
                String method = (String) searchHits[i].getSourceAsMap().get("method");
                String line = (String) searchHits[i].getSourceAsMap().get("line");
                // 目前只处理包含Exception关键字的es日志
                if (StringUtils.contains(message, "Exception")) {
                    String exTypeStr = StringUtils.substringBetween(message, "exception is", ":");
                    String exTypeStr2 = StringUtils.substringBetween(message, ", e=", ":");
                    String ssa = StringUtils.substringAfterLast(message, "Caused by: ");
                    String ex = StringUtils.substringBefore(ssa, "at");
                    if (StringUtils.isNotBlank(exTypeStr)) {
                        // 去掉截取的“[”
                        if (exTypeStr.contains("]")) {
                            exTypeStr = StringUtils.substringBefore(exTypeStr.trim(), "]");
                        }
                        // alreadyComeExTypeSet.add(exTypeStr.trim());
                        map.put(exTypeStr.trim(), 1);
                    } else if (StringUtils.isNotBlank(exTypeStr2)) {
                        if (exTypeStr2.contains("]")) {
                            exTypeStr2 = StringUtils.substringBefore(exTypeStr2.trim(), "]");
                        }
                        map.put(exTypeStr2.trim(), 1);
                        // alreadyComeExTypeSet.add(exTypeStr2.trim());
                    } else if (StringUtils.isNotBlank(ex)) {
                        map.put(ex.trim(), 1);
                    }
                    if (statisticCcuId.containsKey(ccuId)) {
                        Map<String, Integer> map1 = statisticCcuId.get(ccuId);
                        for (Map.Entry<String, Integer> integerEntry : map.entrySet()) {
                            if (map1.containsKey(integerEntry.getKey())) {
                                int value = map1.get(integerEntry.getKey()) + integerEntry.getValue();
                                map1.put(integerEntry.getKey(), value);
                                break;
                            } else {
                                map1.putAll(map);
                            }
                        }
                        Map<String, Integer> tem1 = new ConcurrentHashMap<>();
                        tem1.putAll(map1);
                        if (StringUtils.isNotBlank(ccuId)) {
                            statisticCcuId.put(ccuId, tem1);
                        }
                    } else {
                        Map<String, Integer> tem1 = new ConcurrentHashMap<>();
                        tem1.putAll(map);
                        if (StringUtils.isNotBlank(ccuId)) {
                            statisticCcuId.put(ccuId, tem1);
                        } else {
                            statisticCcuId.put("V_CCU_ID" + UUID.randomUUID(), tem1);
                        }
                    }
                    if (statisticOpcode.containsKey(opcode)) {
                        Map<String, Integer> map2 = statisticOpcode.get(opcode);
                        for (Map.Entry<String, Integer> integerEntry : map.entrySet()) {
                            if (map2.containsKey(integerEntry.getKey())) {
                                int value = map2.get(integerEntry.getKey()) + integerEntry.getValue();
                                map2.put(integerEntry.getKey(), value);
                                break;
                            } else {
                                map2.putAll(map);
                            }
                        }
                        Map<String, Integer> tem1 = new ConcurrentHashMap<>();
                        tem1.putAll(map2);
                        if (StringUtils.isNotBlank(opcode)) {
                            statisticOpcode.put(opcode, tem1);
                        }
                    } else {
                        Map<String, Integer> tem1 = new ConcurrentHashMap<>();
                        tem1.putAll(map);
                        if (StringUtils.isNotBlank(opcode)) {
                            statisticOpcode.put(opcode, tem1);
                        } else {
                            // statisticCcuId.put("V_OPCODE" + UUID.randomUUID(), tem1);
                        }
                    }
                }
            }
        }
        return searchHits;
    }

    public void statSuddenlyIncreaseLogInfo() {
        // 获取昨天的日期
        Date yesterday = DateUtil.offsetDay(new Date(), -1).toJdkDate();
        // 统计昨天现的所有异常类型和对应的次数
        List<AlarmStatDO> statDOS = alarmStatDOMapper
                .selectLogsCount(null, DateUtil.format(yesterday, DatePattern.NORM_DATE_PATTERN));
        // 前天日志
        List<AlarmStatDO> statDOS2 = alarmStatDOMapper
                .selectLogsCount(null, DateUtil.format(DateUtil.offsetDay(yesterday, -1).toJdkDate(), DatePattern.NORM_DATE_PATTERN));
        // 昨天的日志不为空并且比前天的更多
        if (!statDOS.isEmpty() && statDOS.size() - statDOS2.size() >= 500) {
            // 比昨天多500，认为突然增多
            MailDO mailDO = new MailDO();
            mailDO.setTitle("今天Error日志突然增多");
            mailDO.setEmail("1442993929@qq.com,yuanhu@yeah.net");
            mailDO.setContent(String.format("今天Error日志突然增多,比昨天增多了%d,赶紧去kibana查看吧", statDOS.size() - statDOS2.size()));
            mailService.sendTextMail(mailDO);
        }
    }

    public void alarmLsc() {
        // 获取前十分钟
        Date minus_10_minute = DateUtil.offsetMinute(new Date(), -10).toJdkDate();
        // 统计前十分钟之内的异常信息
        List<AlarmStatDO> statDOS = alarmStatDOMapper
                .selectLogsBetween(null, DateUtil.format(minus_10_minute, DatePattern.NORM_DATE_PATTERN),
                        DateUtil.format(new Date(), DatePattern.NORM_DATE_PATTERN));
        for (AlarmStatDO statDO : statDOS) {
            if (statDO.getExTypeName().contains("ConnectException")) {
                MailDO mailDO = new MailDO();
                mailDO.setTitle("ConnectException");
                mailDO.setEmail("1442993929@qq.com,yuanhu@yeah.net");
                mailDO.setContent("检测到" + DateUtil.format(statDO.getCreateDateTime(), DatePattern.NORM_DATETIME_PATTERN)
                        + "出现ConnectException，请去es查看具体错误信息！！！");
                mailService.sendTextMail(mailDO);
            }
        }

    }

    /**
     * 记录主机推送消息
     */
    public void recordPushMsg() throws IOException {
        DateTime startTime = DateUtil.date();
        final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(1L));
        SearchRequest searchRequest = new SearchRequest("filebeat*");
        searchRequest.scroll(scroll);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(QueryBuilders.boolQuery()
                // .must(QueryBuilders.matchAllQuery())
                // .must(QueryBuilders.matchQuery("message", "主机同步失败"))
                // .must(QueryBuilders.matchQuery("ccuid", "CCU_08069"))
                // 模糊查询
                .must(QueryBuilders.multiMatchQuery("push event send msg", "message"))
                // 范围查询
                .filter(QueryBuilders.rangeQuery("@timestamp")
                        .gte(DateUtil.offsetMinute(startTime.toJdkDate(), -2)
                                // 秒置零
                                .setField(DateField.SECOND, 0)
                                // 毫秒置零
                                .setField(DateField.MILLISECOND, 0)
                                .getTime())
                        .lte(DateUtil.offsetMinute(startTime.toJdkDate(), -1)
                                // 秒置零
                                .setField(DateField.SECOND, 59)
                                // 毫秒置零
                                .setField(DateField.MILLISECOND, 999)
                                .getTime()))
        // 排序
        )
                .sort(SortBuilders.fieldSort("@timestamp").order(SortOrder.DESC));
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = esClient.search(searchRequest, RequestOptions.DEFAULT);
        String scrollId = searchResponse.getScrollId();
        SearchHit[] searchHits = searchResponse.getHits().getHits();
        while (searchHits != null && searchHits.length > 0) {
            processSearchHits(searchHits);
            SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
            scrollRequest.scroll(scroll);
            searchResponse = esClient.scroll(scrollRequest, RequestOptions.DEFAULT);
            scrollId = searchResponse.getScrollId();
            searchHits = searchResponse.getHits().getHits();
        }
        ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
        clearScrollRequest.addScrollId(scrollId);
        ClearScrollResponse clearScrollResponse = esClient.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        boolean succeeded = clearScrollResponse.isSucceeded();
        if (succeeded) {
            log.info("scroll 清理成功");
        } else {
            log.warn("scroll 清理失败");
        }
        DateTime endTime = DateUtil.date();
        log.info("记录主机推送消息完成，耗时：{}毫秒", DateUtil.between(startTime.toJdkDate(), endTime.toJdkDate(), DateUnit.MS));
    }

    /**
     * 处理SearchHits结果集
     *
     * @return
     */
    private SearchHit[] processSearchHits(SearchHit[] searchHits) {
        if (searchHits != null && searchHits.length > 0) {
            for (int i = 0; i < searchHits.length; i++) {
                // 异常信息
                String message = (String) searchHits[i].getSourceAsMap().get("message");
                String ccuId = (String) searchHits[i].getSourceAsMap().get("ccuid");
                String opcode = (String) searchHits[i].getSourceAsMap().get("opcode");
                String pushMsgStr = StringUtils.trim(StringUtils.substringBetween(message, "push event send msg", "to:"));
                // 解析pushMsg
                /*
                 * { "pushCcuId": "79fa3ba4-5eda-4d83-9bc9-0786132a8cf7", "pushMsgType": "DeviceStatusChangedPushEvent", "pushMsgId":
                 * "M_5a6ec7b2-0099-41e1-aa9b-01f0ad641dd3", "pushMsgTimestamp": 1613958111690, "pushMsg": { "id": 29, "type":
                 * "KONKE_ZIGBEE_CHOPIN_10A_SOCKET", "status": { "on": true, "load": true, "onlineState": "ONLINE" } } }
                 */
                if (StringUtils.isBlank(pushMsgStr)) {
                    continue;
                }
                ONode o = ONode.loadStr(pushMsgStr);
                // 得到所有的书
                String pushMsgType = o.get("pushMsgType").getString();
                String type = o.select("$.pushMsg.type").getString();
                String pushMsg = o.select("$.pushMsg").toJson();
                // 得到所有的书名
                PushMsgDO pushMsgDO = new PushMsgDO();
                pushMsgDO.setMsg(pushMsg);
                pushMsgDO.setType(type);
                pushMsgDO.setPushMsgType(pushMsgType);
                pushMsgDO.setCcuId(ccuId);
                pushMsgDO.setOpcode(opcode);
                // 查询,除了DeviceStatusChangedPushEvent,其它事件不需要重复
                if ("DeviceStatusChangedPushEvent".equals(type)) {
                    Example example = new Example(PushMsgDO.class);
                    Example.Criteria criteria = example.createCriteria();
                    criteria.andEqualTo("type", type);
                    List<PushMsgDO> pushMsgDOS = pushMsgDOMapper.selectByExample(example);
                    if (CollUtil.isEmpty(pushMsgDOS)) {
                        pushMsgDOMapper.insert(pushMsgDO);
                    }
                } else {
                    Example example = new Example(PushMsgDO.class);
                    Example.Criteria criteria = example.createCriteria();
                    criteria.andEqualTo("pushMsgType", pushMsgType);
                    List<PushMsgDO> pushMsgDOS = pushMsgDOMapper.selectByExample(example);
                    if (CollUtil.isEmpty(pushMsgDOS)) {
                        pushMsgDOMapper.insert(pushMsgDO);
                    }
                }

            }
        }
        return searchHits;
    }
}
